package com.softdev.system.demo.handler;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * MyWebSocketHandler
 * WebSocket处理器，处理websocket连接相关
 * @author zhengkai.blog.csdn.net
 * @date 2019-06-12
 */
@Slf4j
public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
	/** 模型数据缓存 */
	private static final Map<Object, List<Map<String, Object>>> modelCache = Maps.newConcurrentMap();
	/** 模型接收的指令缓存 */
	private static final Map<Object, Object> modelCmdCache = Maps.newConcurrentMap();
	
	/** 模型接收的指令缓存 */
	private static final Map<Object, Object> modelRealTimeThread = Maps.newConcurrentMap();
	
	/**
     * IO密集型任务  = 一般为2*CPU核心数（常出现于线程中：数据库数据交互、文件上传下载、网络数据传输等等）
     * CPU密集型任务 = 一般为CPU核心数+1（常出现于线程中：复杂算法）
     * 混合型任务  = 视机器配置和复杂度自测而定
     */
    private static int corePoolSize = Runtime.getRuntime().availableProcessors();
    /**
     * public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,
     *                           TimeUnit unit,BlockingQueue<Runnable> workQueue)
     * corePoolSize用于指定核心线程数量
     * maximumPoolSize指定最大线程数
     * keepAliveTime和TimeUnit指定线程空闲后的最大存活时间
     * workQueue则是线程池的缓冲队列,还未执行的线程会在队列中等待
     * 监控队列长度，确保队列有界
     * 不当的线程池大小会使得处理速度变慢，稳定性下降，并且导致内存泄露。如果配置的线程过少，则队列会持续变大，消耗过多内存。
     * 而过多的线程又会 由于频繁的上下文切换导致整个系统的速度变缓——殊途而同归。队列的长度至关重要，它必须得是有界的，这样如果线程池不堪重负了它可以暂时拒绝掉新的请求。
     * ExecutorService 默认的实现是一个无界的 LinkedBlockingQueue。
     */
    private static ThreadPoolExecutor executor  = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(1000));
	
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端建立连接，通道开启！");

        //添加到channelGroup通道组
        MyChannelHandlerPool.channelGroup.add(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端断开连接，通道关闭！");
        //添加到channelGroup 通道组
        MyChannelHandlerPool.channelGroup.remove(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //首次连接是FullHttpRequest，处理参数 by zhengkai.blog.csdn.net
        if (null != msg && msg instanceof FullHttpRequest) {
            FullHttpRequest request = (FullHttpRequest) msg;
            String uri = request.uri();

            Map paramMap=getUrlParams(uri);
            log.info("接收到的参数是："+JSON.toJSONString(paramMap));
            //如果url包含参数，需要处理
            if(uri.contains("?")){
                String newUri=uri.substring(0,uri.indexOf("?"));
                log.info("new Uri: {}", newUri);
                request.setUri(newUri);
            }

        }else if(msg instanceof TextWebSocketFrame){
            //正常的TEXT消息类型
            TextWebSocketFrame frame=(TextWebSocketFrame)msg;
            log.info("客户端收到服务器数据：" +frame.text());
            Map mapJson = JSON.parseObject(frame.text(), Map.class);
            Object modelId = mapJson.get("modelId");
            Object work = mapJson.get("work");
            if ("query".equals(work)) {
            	//开始工作
            	ChannelFuture channelFuture = sendMessage(ctx, "收到任务：" + frame.text());
            	channelFuture.addListener((future) -> {
                    log.info("开始执行异步任务。");
                    List<Map<String, Object>> list = getCurrentBizData(modelId);
                	modelCache.put(modelId, list);
                	sendMessage(ctx, modelId + " 模型数据： " + JSON.toJSONString(list));
                	monitor(ctx, mapJson);
            	});
            	log.info("接下来执行异步任务。");
            	
            } else {
            	monitor(ctx, mapJson);
            }
            //sendAllMessage(frame.text());
        }
        super.channelRead(ctx, msg);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {

    }

    /**
     * 向全部通道群发消息
     * @param message
     */
    private void sendAllMessage(String message){
        //收到信息后，群发给所有channel
        MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
    }
    /**
     * 向单个通道发消息
     * @param ctx
     * @param message
     * @return
     */
    private ChannelFuture sendMessage(ChannelHandlerContext ctx, String message){
        //收到信息后，群发给所有channel
        ChannelFuture channelFuture = ctx.channel().writeAndFlush( new TextWebSocketFrame(message));
        return channelFuture;
    }

    private static Map getUrlParams(String url){
        Map<String,String> map = new HashMap<>();
        url = url.replace("?",";");
        if (!url.contains(";")){
            return map;
        }
        if (url.split(";").length > 0){
            String[] arr = url.split(";")[1].split("&");
            for (String s : arr){
                String key = s.split("=")[0];
                String value = s.split("=")[1];
                map.put(key,value);
            }
            return  map;

        }else{
            return map;
        }
    }
    
    private List<Map<String, Object>> getCurrentBizData(Object modelId) {
    	List<Map<String, Object>> list = Lists.newArrayList();
    	Map<String, Object> map = Maps.newHashMap();
    	Map<String, Object> map1 = Maps.newHashMap();
    	Map<String, Object> map2 = Maps.newHashMap();
    	list.add(map);
    	list.add(map1);
    	list.add(map2);
    	
    	map.put("name", "张三" + modelId);
    	map.put("pwd", "123456");
    	map1.put("name", "李四" + modelId);
    	map1.put("pwd", "123456");
    	map2.put("name", "王五" + modelId);
    	map2.put("pwd", "123456");
    	return list;
    }
    private void monitor(ChannelHandlerContext ctx, Map modelJson) {
    	Object modelId = modelJson.get("modelId");
    	Object realtime = modelJson.get("realtime");
    	log.info("{} 实时监控：{}", modelId, realtime);
    	modelCmdCache.put(modelId, realtime);
    	//实时监控
    	if ("start".equals(realtime)) {
    		Object thread = modelRealTimeThread.get(modelId);
    		if (thread == null || "inactive".equals(thread)) {
    			executor.submit(()->{
    				while (!"stop".equals(modelCmdCache.get(modelId))) {
    					log.info("校验业务数据变化 {}", modelId);
    					List<Map<String, Object>> listCached = modelCache.get(modelId);
    					//获取当前业务数据
    					List<Map<String, Object>> list2 = getCurrentBizData(modelId);
    					if (listCached.size() != list2.size()) {
    						sendMessage(ctx, modelId + " 数据产生变化，新业务数据大小：" + list2.size());
    						modelCache.put(modelId, list2);
    					}
    					try {
    						Thread.sleep(10000);
    					} catch (Exception e) {
    						e.printStackTrace();
    					}
    				}
    				log.info("{} 实时监控线程执行结束。", modelId);
    				modelRealTimeThread.put(modelId, "inactive");//实时监控线程清空
    				sendMessage(ctx, modelId + " 成功关闭实时监控 ");
    			});
    			modelRealTimeThread.put(modelId, "active");
    			sendMessage(ctx, modelId+" 成功启动实时监控 ");
    		} else {
    			sendMessage(ctx, modelId+" 已运行实时监控，不需要重复运行");
    		}
    	} else if ("stop".equals(realtime)){
    		Object thread = modelRealTimeThread.get(modelId);
    		if (thread == null || "inactive".equals(thread)) {
    			sendMessage(ctx, modelId+" 已关闭实时监控，不需要重复关闭");
    		} else {
    			sendMessage(ctx, modelId+" 发送关闭实时监控命令");
    		}
    	}    
    }
}